

package org.example.flinksql.test;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.example.flinksql.test.custom.*;

/**
 * 验证单表数据同步
 *
 * 总数据量：50w+数据
 * 表结构一致时数据同步sink单并行度情况下同步数据耗时12min
 * 表结构不一致、加入转换逻辑、sink单并行度情况下同步数据耗时31min
 */
public class SyncAndTransPartners {

	public static void main(String[] args) throws Exception {
		EnvironmentSettings bsSettings = EnvironmentSettings.newInstance()
				.useBlinkPlanner().inStreamingMode().build();
		TableEnvironment tbEnv = TableEnvironment.create(bsSettings);

		int limit = 10;
		if (null != args && 0 != args.length) {
			String limitStr = args[0];
			limit = Integer.valueOf(limitStr);
		}


		String srcHost = "localhost";
		String srcPort = "3306";
		String srcUsername = "root";
		String srcPassword = "root";
		String srcDb = "db_enterprise_0";
		String srcTb = "t_partners_0";
		String srcTbStruct = "id BIGINT," +
				"eid STRING," +
				"entity_type STRING," +
				"seq_no INT," +
				"category STRING," +
				"stock_name STRING," +
				"stock_id STRING," +
				"stock_type STRING," +
				"stock_percent STRING," +
				"identify_type STRING," +
				"identify_no STRING," +
				"total_should_capi STRING," +
				"should_capi_items STRING," +
				"total_real_capi STRING," +
				"real_capi_items STRING," +
				"tag STRING," +
				"ex_id STRING," +
				"country STRING," +
				"row_update_time TIMESTAMP";

		String dstHost = "localhost";
		String dstPort = "3306";
		String dstUsername = "flinksql";
		String dstPassword = "w0c53w0c53";
		String dstDb = "db_test";
		String dstTb = "t_partners";
		String dstTbStruct = "id BIGINT," +
				"eid STRING," +
				"entity_type INT," +
				"seq_no INT," +
				"stock_name STRING," +
				"stock_id STRING," +
				"stock_type INT," +
				"stock_percent DECIMAL(10,6)," +
				"identify_type INT," +
				"identify_no STRING," +
				"total_should_capi DECIMAL(20,6)," +
				"should_invest_type STRING," +
				"should_capi_date DATE," +
				"shoud_capi_value DECIMAL(20,6)," +
				"total_real_capi DECIMAL(20,6)," +
				"real_invest_type STRING," +
				"real_capi_date DATE," +
				"real_capi_value DECIMAL(20,6)," +
				"country STRING," +
				"created_time TIMESTAMP," +
				"row_update_time TIMESTAMP";


		tbEnv.executeSql(String.format("CREATE TABLE t_src (%s) WITH (" +
				" 'connector' = 'jdbc', " +
				" 'url' = 'jdbc:mysql://%s:%s/%s" +
				"?useSSL=false&useUnicode=true&characterEncoding=UTF-8', " +
				" 'table-name' = '%s', " +
				" 'username' = '%s', " +
				" 'password' = '%s' " +
				")", srcTbStruct, srcHost, srcPort, srcDb, srcTb, srcUsername, srcPassword));

		tbEnv.executeSql(String.format("CREATE TABLE t_dst (%s) WITH (" +
				" 'connector' = 'jdbc', " +
				" 'url' = 'jdbc:mysql://%s:%s/%s" +
				"?useSSL=false&useUnicode=true&characterEncoding=UTF-8', " +
				" 'table-name' = '%s', " +
				" 'username' = '%s', " +
				" 'password' = '%s' " +
//				" 'sink.parallelism' = '3' " +
				")", dstTbStruct, dstHost, dstPort, dstDb, dstTb, dstUsername, dstPassword));

		tbEnv.createTemporaryFunction("transDate", TransDateFunc.class);
		tbEnv.createTemporaryFunction("transDatetime", TransDatetimeFunc.class);
		tbEnv.createTemporaryFunction("transInteger", TransIntegerFunc.class);
		tbEnv.createTemporaryFunction("transDecimal", TransDecimalFunc.class);

		tbEnv.createTemporaryFunction("transET", TransEntityTypeFunc.class);
		tbEnv.createTemporaryFunction("transIdT", TransIdentifyTypeFunc.class);
		tbEnv.createTemporaryFunction("transInT", TransInvestTypeFunc.class);
		tbEnv.createTemporaryFunction("transST", TransStockTypeFunc.class);

		tbEnv.createTemporaryFunction("TransTC", TransTotalCapiFunc.class);

		tbEnv.createTemporaryFunction("TransSCD", TransShouldCapiDateFunc.class);
		tbEnv.createTemporaryFunction("TransSCV", TransShouldCapiValueFunc.class);
		tbEnv.createTemporaryFunction("TransRCD", TransRealCapiDateFunc.class);
		tbEnv.createTemporaryFunction("TransRCV", TransRealCapiValueFunc.class);


		Table table = tbEnv.sqlQuery(String.format("select id,eid,transET(entity_type) as entity_type,seq_no," +
				"stock_name,stock_id,transST(stock_type) as stock_type,transDecimal(stock_percent) as stock_percent," +
				"transIdT(identify_type) as identify_type,identify_no,TransTC(total_should_capi) as total_should_capi," +
				"transInT(should_capi_items) as should_invest_type,TransSCD(should_capi_items) as should_capi_date," +
				"TransSCV(should_capi_items) as shoud_capi_value,TransTC(total_real_capi) as total_real_capi," +
				"transInT(real_capi_items) as real_invest_type,TransRCD(real_capi_items) as real_capi_date," +
				"TransRCV(real_capi_items) as real_capi_value,country," +
				"row_update_time as create_time, row_update_time" +
				" from t_src limit %d", limit));
		table.executeInsert("t_dst");
	}
}
